package com.galudisu

import java.nio.file.StandardOpenOption._

import akka.actor.ActorSystem
import akka.stream.scaladsl.{FileIO, RunnableGraph, Sink, Source}
import akka.stream.{ActorMaterializer, IOResult}
import akka.util.ByteString
import com.typesafe.config.ConfigFactory

import scala.concurrent.Future
/**
  * @author galudisu
  */
object StreamingCopy extends App {

  val config = ConfigFactory.load()
  val maxLine = config.getInt("log-stream-processor.max-line")

  if(args.length != 2) {
    System.err.println("Provide args: input-file output-file")
    System.exit(1)
  }

  val inputFile = FileArg.shellExpanded(args(0))
  val outputFile = FileArg.shellExpanded(args(1))

  // imports for the StreamingCopy app
  // `Source`
  val source:Source[ByteString,Future[IOResult]] = FileIO.fromPath(inputFile)
  // `Sink`
  val sink:Sink[ByteString,Future[IOResult]] = FileIO.toPath(outputFile,Set(CREATE,WRITE,APPEND))

  // `Source` to `Sink`，`to`方法将Source 和 Sink组合到一个 module 中
  // `ActorMaterializer`根据module的连接情况，对publishers进行subscribes
  // FilePublisher 读取 ByteStrings
  val runnableGraph: RunnableGraph[Future[IOResult]] = source.to(sink)

  // The materializer eventually creates actors that execute the graph
  implicit val system = ActorSystem()
  implicit val ec = system.dispatcher
  implicit val materializer = ActorMaterializer()

  // Running the graph returns a Future[IOResult]; in this case, the IOResult contains a count of bytes read from the source.
  runnableGraph.run().foreach { result =>
    println(s"${result.status}, ${result.count} bytes read.")
    system.terminate()
  }

  // These are just examples, they are not run as part of StreamingCopy

  import akka.Done
  import akka.stream.scaladsl.Keep

  // `to`方法默认返回`Source`定义的auxiliary value，
  // 即`Source`-`to`-`Sink`，默认根据左边left，
  // `Source`的auxiliary value作为返回类型，下面是它的具体返回类型的一些实现
  val graphLeft: RunnableGraph[Future[IOResult]] =
    source.toMat(sink)(Keep.left)         // Keeps the IOResult of reading the file
  val graphRight: RunnableGraph[Future[IOResult]] =
    source.toMat(sink)(Keep.right)        // Keeps the IOResult of writing the file
  val graphBoth: RunnableGraph[(Future[IOResult], Future[IOResult])] =
    source.toMat(sink)(Keep.both)         // Keeps both
  val graphCustom: RunnableGraph[Future[Done]] =
    source.toMat(sink) { (l, r) =>
      Future.sequence(List(l,r)).map(_ => Done) // Custom function just indicates the stream is done
    }


}
